package com.heima.kafka.produce;


import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @BelongsProject: heima-leadnews
 * @BelongsPackage: com.heima.kafka.produce
 * @Author: wangjian
 * @CreateTime: 2022-07-08  17:29
 * @Description: TODO
 * @Version: 1.0
 */

public class ProducerTest {
    public static void main(String[] args) {
        // 设置属性
        Properties props = new Properties();
        // 指定连接的kafka服务器的地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.85.143:9092");
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        // 构建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props, new StringSerializer(), new StringSerializer());
        // 定义kafka主题
        String topic = "TextLinesTopic";
        // 调用生产者发送消息
        String message = "tom jerry jack tom tom jack";
        // 构建消息
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, message);
        // 同时发送key和value
        record = new ProducerRecord<String, String>(topic, "10010", message);
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception!=null){
//                    没有发送成功
                    exception.printStackTrace();
                }
                System.out.println(metadata.offset());
            }
        });

        // 释放连接
        producer.close();
    }
}
